package com.luoqifei;

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.KStreamBuilder;
import org.apache.kafka.streams.kstream.KTable;

import java.util.*;
import java.util.concurrent.CountDownLatch;

public class Pipe {
    public static void main(String[] args) {
        Properties props = new Properties();
        //set the stream application unique identifier
        props.put(StreamsConfig.APPLICATION_ID_CONFIG,"stream-pipe");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,"localhost:9092");
        props.put(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG,10*1024*1024L);//10Mb record cache size 10Mb
        props.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG,10000);
        props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG,1);//exactly once
        props.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG,1);
        //serialization and deserialization libraries
        props.put(StreamsConfig.KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        props.put(StreamsConfig.VALUE_SERDE_CLASS_CONFIG,Serdes.String().getClass());

        final KStreamBuilder builder = new KStreamBuilder();
        KStream<String,String> source = builder.stream("test");//source topics
        KTable<String,String> kTable = builder.table(",","");
        //word count
//        source.flatMapValues(value -> Arrays.asList(value.toLowerCase(Locale.getDefault()).split("\\W+")))
//                .groupBy((key, value) -> value)//count by key,value is times
//                .count("Counts")//state state name counts,this store (also call k table) can query in real-time
//                .to(Serdes.String(), Serdes.Long(), "streams-wordcount-output");//save this k table log to topic



        source.to("kstream-output");//k stream output

        final KafkaStreams streams = new KafkaStreams(builder,props);


        //get ctrl+c,shutdown this k stream
        final CountDownLatch latch = new CountDownLatch(1);
        Runtime.getRuntime().addShutdownHook(new Thread("stream-shutdown"){
            @Override
            public void run(){
                streams.close();
                latch.countDown();
            }
        });

        streams.start();
        try {
            latch.await();
        } catch (InterruptedException e) {
            System.exit(1);
        }
        System.exit(0);
    }
}
